feat(streaming): arrangement backfill #10266

kwannoel · 2023-06-09T10:06:01Z

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

RFC: https://www.notion.so/risingwave-labs/Arrangement-Backfill-31d4d0044726451c9cc77dc84a0f8bb2

We implement the arrangement backfill executor in this PR.
Integration tests are difficult to write for it, since CreateMviewProgress needs to be mocked as well, which does not seem trivial to me.

Since this is not used yet, I think it is fine to just do a code review of the corresponding changes to the code base before merging. I will defer frontend changes to later to avoid too large a PR.

It will be tested later when frontend support is added too. When that happens I will test it e2e with existing e2e test + adding some scale and recovery tests specific to arrangement backfill.

It will also be benchmarked against backfill then.

Comparison to backfill:

StateTable is needed for replication. So it uses StateTable instead of StorageTable. Later frontend will need to take care of instantiating the table, making sure it is of type is_replicated. We should add an assertion for ArrangementBackfill::new when that's done.
When we receive barrier, we have to flush the changes to state_table, to ensure it is replicated, before any new snapshot read.
Because this involves a write to StateTable, this also means we have to break out of the backfill stream loop first. This is so we drop the immutable reference to StateTable, which is used to read snapshot.
As a result we shift the entire barrier processing logic to outside the backfill stream loop.
It makes the logic a little convoluted, I welcome ideas to simplify it further. Refactoring it into a function (process_barrier) does not seem ideal as well, since it covers many variables >= 10.

Changes to backfill state persistence:

In this PR we track progress per vnode.
This is so we can support scaling for it, which is the main usecase (independently scale backfill with upstream mv).

Changes to snapshot read:

Because backfill could now have mix of states as a result of scaling (some vnodes complete, others in progress / not-started)
This means our snapshot read logic will skip completed vnodes, start from different offsets / no offset for in-progress and not-started vnodes respectively.

Changes to update processing (cr @hzxa21):

Skip a row if it belongs to a not-yet-scanned vnode or a scanning vnode with smaller current_pos
Emit a row to downstream if it belongs to a fully-scanned vnode.
Emit a row to downstream if it belongs to a scanning vnode with larger current_pos

Other refactoring:

Reuse interfaces from backfill, so we refactor backfill.rs -> backfill/no_shuffle_backfill.rs
and common utilities to backfill/utils.rs

Here are further changes to existing interfaces (OUTDATED, IGNORE THIS SECTION):

These changes are mainly for state table to support reads (previously we use interface of StorageTable, which already supported our read pattern). Our read pattern is different, so we need additional interfaces.
Implement state table interface for iterating through state_table by vnodes.
This is supported by the iter_all_vnodes_with_pk_range.
Support ordered iteration of rows. This is done by duplicating merge_sort implementation from batch. Ideally I would like to reuse the original merge_sort there, but there are some difficulties as described in the docstring.
Also support iter by chunk, via collect_data_chunk. I managed to refactor it so it could be reused, but can't quite keep this method in a trait, due to orphan instance rules.
I kept the TableIter interface, since several locations use its next_row method still. It can be removed next time.

Regarding persisting arrangement backfill state:

Currently I keep the logic from backfill used to persist the state.
This is because it seems to require minimal logic (just need to update pk_in_output_indices) to maintain it.
If it causes trouble after arrangement_backfill is implemented e2e, I will remove it.
Otherwise it makes sense to keep it, since it will be much harder to add back later.

Checklist For Contributors

I have written necessary rustdoc comments
I have added necessary unit tests and integration tests
I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
I have demonstrated that backward compatibility is not broken by breaking changes and created issues to track deprecated features to be removed in the future. (Please refer to the issue)
All checks passed in ./risedev check (or alias, ./risedev c)

Checklist For Reviewers

I have requested macro/micro-benchmarks as this PR can affect performance substantially, and the results are shown.

Documentation

My PR DOES NOT contain user-facing changes.

Click here for Documentation

Types of user-facing changes

Please keep the types that apply to your changes, and remove the others.

Installation and deployment
Connector (sources & sinks)
SQL commands, functions, and operators
RisingWave cluster configuration changes
Other (please specify in the release note below)

Release note

codecov · 2023-06-15T09:06:08Z

Codecov Report

Merging #10266 (9cc6ad2) into main (5443a1e) will decrease coverage by 0.09%.
The diff coverage is 13.75%.

@@            Coverage Diff             @@
##             main   #10266      +/-   ##
==========================================
- Coverage   70.34%   70.26%   -0.09%     
==========================================
  Files        1274     1276       +2     
  Lines      219041   219308     +267     
==========================================
+ Hits       154085   154097      +12     
- Misses      64956    65211     +255

Flag	Coverage Δ
rust	`70.26% <13.75%> (-0.09%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
src/batch/src/executor/row_seq_scan.rs	`21.49% <ø> (ø)`
src/storage/src/table/batch_table/storage_table.rs	`86.81% <0.00%> (ø)`
...ream/src/executor/backfill/arrangement_backfill.rs	`0.00% <0.00%> (ø)`
...tream/src/executor/backfill/no_shuffle_backfill.rs	`0.00% <0.00%> (ø)`
src/stream/src/executor/backfill/utils.rs	`0.00% <0.00%> (ø)`
src/stream/src/executor/batch_query.rs	`82.22% <ø> (ø)`
src/stream/src/executor/mod.rs	`51.92% <ø> (ø)`
src/stream/src/common/table/state_table.rs	`88.53% <75.00%> (-0.22%)`	⬇️
src/storage/src/table/merge_sort.rs	`88.67% <100.00%> (ø)`
src/storage/src/table/mod.rs	`83.01% <100.00%> (+1.01%)`	⬆️
... and 1 more

... and 7 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

kwannoel · 2023-07-04T16:40:44Z

Have updated the implementation, it should match the proposal from @hzxa21 now: #10266 (comment).

Will leave some comments on certain parts which may need further review.

Will do a second pass tomorrow as well.

src/stream/src/executor/backfill/arrangement_backfill.rs

chenzl25 · 2023-07-06T06:43:54Z

The new snapshot_read_per_vnode LGTM.

hzxa21

Nicely done. Thanks for the PR! Left some comments.

Rest LGTM!

hzxa21 · 2023-07-07T11:45:51Z

src/stream/src/executor/backfill/utils.rs

+    debug_assert!(!state_table.vnode_bitmap().is_empty());
+    let vnodes = state_table.vnodes().iter_vnodes();
+    let mut result = Vec::with_capacity(state_table.vnodes().len());
+    for vnode in vnodes {


Since get_progress_per_vnode is called when we need to init the backfill state, cache of state table is highly likely to be empty and each get_row will be an actual S3 I/O. How about using futures::future::try_join_all to issue state_table.get_row concurrently to reduce overall latency?

I assumed rust compiler be able perform each get_row concurrently, even within a for loop. Guess I should build a small toy program to test it out.

I will follow your suggestion to be safe.

EDIT: Thinking about it again, I guess if we leave it to rust runtime, it may not schedule all get_row concurrently, e.g. stagger them? So try_join_all will enforce it.

Updated in 9cc6ad2

hzxa21 · 2023-07-07T11:47:24Z

src/stream/src/executor/backfill/arrangement_backfill.rs

+            .iter()
+            .all(|(_, p)| *p == BackfillProgressPerVnode::Completed);
+        if is_completely_finished {
+            assert!(!first_barrier.is_newly_added(self.actor_id));


Isn't this also true when is_completely_finished == false?

If is_completely_finished == false, this assertion can be false as well.
For instance if all progress is BackfillProgressPerVnode::NotStarted. Which could mean it is from new streaming graph, and this assertion will be false.

add arrangement backfill executor

a211291

github-actions bot added the type/feature Type: New feature. label Jun 9, 2023

kwannoel added 13 commits June 9, 2023 18:28

iter over pk bounds of state table

72a50c6

add ordered iter stub

6e573c2

add reference to schema

8c6dc84

duplicate collect_data_chunk for now

d6bc636

enable collect_data_chunk

f9c688d

add back instrumentation

022710f

checkpoint the work on merge_sort

2f37d1b

make compile

6f8a97e

tmp commit

f55d375

hack

098bedf

ok works lol

2b827c5

finally last part of merge sort

c9ed209

make it compile with arrangement backfill too

57f1edd

github-actions bot added the user-facing-changes Contains changes that are visible to users label Jun 15, 2023

kwannoel added 6 commits June 15, 2023 15:31

refactor merge_sort to separate module

7ea8503

unified collect_chunk somewhat

b7b5173

make the diff better

b540eb7

remove outdated comment

7fdcb2f

fix outdated comment

faf7c50

risedev cf

cbb2127

kwannoel changed the title ~~feat(streaming): arrangement backfill~~ feat(streaming): arrangement backfill [IGNORE ME, WIP] Jun 15, 2023

kwannoel marked this pull request as ready for review June 15, 2023 08:34

some more refactoring + run on ci to make sure nothing broke

120bb70

kwannoel added 3 commits June 16, 2023 11:57

remove outdated code

5b3e034

borrow chunk when flushing to downstream

bd2cf4a

finish implementing buffered chunk logic

16ece5b

kwannoel force-pushed the kwannoel/arrangement-backfill branch from 83c4e01 to 16ece5b Compare June 16, 2023 09:37

kwannoel added 11 commits July 4, 2023 19:48

workaround current_pos_map reference, by cloning it

d522787

add FIXME

da8fffb

update persist_data per vnode

5e4a5c2

naming + init backfill state

3558a0d

refactor to snapshot_read with backfill_state

12d7d93

update progress to backfill_state

df4942a

update mark_chunk_ref_by_vnode to use backfill_state

55d26de

interim commit: persist_state_per_vnode

56d853d

use persist_state_per_vnode

259c278

clean

d1ff383

minor

c3edb09

kwannoel commented Jul 4, 2023

View reviewed changes

src/stream/src/executor/backfill/arrangement_backfill.rs Show resolved Hide resolved

kwannoel commented Jul 4, 2023

View reviewed changes

src/stream/src/executor/backfill/arrangement_backfill.rs Show resolved Hide resolved

kwannoel commented Jul 4, 2023

View reviewed changes

src/stream/src/executor/backfill/arrangement_backfill.rs Show resolved Hide resolved

github-actions bot added the user-facing-changes Contains changes that are visible to users label Jul 4, 2023

kwannoel removed the user-facing-changes Contains changes that are visible to users label Jul 4, 2023

kwannoel marked this pull request as ready for review July 4, 2023 16:50

github-actions bot added the user-facing-changes Contains changes that are visible to users label Jul 4, 2023

kwannoel requested a review from hzxa21 July 7, 2023 04:36

hzxa21 approved these changes Jul 7, 2023

View reviewed changes

use try_join_all

9cc6ad2

kwannoel requested a review from hzxa21 July 7, 2023 18:02

kwannoel added this pull request to the merge queue Jul 10, 2023

Merged via the queue into main with commit 019e7b4 Jul 10, 2023

kwannoel deleted the kwannoel/arrangement-backfill branch July 10, 2023 03:34

kwannoel removed the user-facing-changes Contains changes that are visible to users label Jul 10, 2023

xxhZs mentioned this pull request Jul 20, 2023

Bug(sink): Create iceberg sink error. #11091

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

feat(streaming): arrangement backfill #10266

feat(streaming): arrangement backfill #10266

Uh oh!

kwannoel commented Jun 9, 2023 •

edited

Loading

Uh oh!

codecov bot commented Jun 15, 2023 •

edited

Loading

Uh oh!

kwannoel commented Jul 4, 2023 •

edited

Loading

Uh oh!

Uh oh!

Uh oh!

Uh oh!

chenzl25 commented Jul 6, 2023

Uh oh!

hzxa21 left a comment

Uh oh!

hzxa21 Jul 7, 2023

Uh oh!

kwannoel Jul 7, 2023 •

edited

Loading

Uh oh!

kwannoel Jul 7, 2023 •

edited

Loading

Uh oh!

hzxa21 Jul 7, 2023

Uh oh!

kwannoel Jul 7, 2023 •

edited

Loading

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

7 participants

feat(streaming): arrangement backfill #10266

feat(streaming): arrangement backfill #10266

Uh oh!

Conversation

kwannoel commented Jun 9, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What's changed and what's your intention?

Checklist For Contributors

Checklist For Reviewers

Documentation

Types of user-facing changes

Release note

Uh oh!

codecov bot commented Jun 15, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

kwannoel commented Jul 4, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

chenzl25 commented Jul 6, 2023

Uh oh!

hzxa21 left a comment

Choose a reason for hiding this comment

Uh oh!

hzxa21 Jul 7, 2023

Choose a reason for hiding this comment

Uh oh!

kwannoel Jul 7, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kwannoel Jul 7, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hzxa21 Jul 7, 2023

Choose a reason for hiding this comment

Uh oh!

kwannoel Jul 7, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

7 participants

kwannoel commented Jun 9, 2023 •

edited

Loading

codecov bot commented Jun 15, 2023 •

edited

Loading

kwannoel commented Jul 4, 2023 •

edited

Loading

kwannoel Jul 7, 2023 •

edited

Loading

kwannoel Jul 7, 2023 •

edited

Loading

kwannoel Jul 7, 2023 •

edited

Loading